package order

import (
	"bytes"
	"encoding/json"
	"errors"
	"io/ioutil"
	"learning/gooop/saga/mqs/logger"
	"learning/gooop/saga/mqs/models"
	"net/http"
	"sync"
	"sync/atomic"
	"time"
)

type tSaleOrderService struct {
	rwmutex *sync.RWMutex
	orders map[string]*SaleOrder
	bMQReady bool
	publishQueue chan *SaleOrder
}

func newSaleOrderService() ISaleOrderService {
	it := new(tSaleOrderService)
	it.init()
	return it
}

func (me *tSaleOrderService) init() {
	me.rwmutex = new(sync.RWMutex)
	me.orders = make(map[string]*SaleOrder)
	me.bMQReady = false
	me.publishQueue = make(chan *SaleOrder, gMQMaxQueuedMsg)

	go func() {
		time.Sleep(100*time.Millisecond)
		go me.beginSubscribeMQ()
		go me.beginPublishMQ()
	}()
}

func (me *tSaleOrderService) beginSubscribeMQ() {
	expireDuration := int64(1 * time.Hour)
	subscribeDuration := 20 * time.Minute
	pauseDuration := 3*time.Second
	lastSubscribeTime := int64(0)

	for {
		now := time.Now().UnixNano()
		if now - lastSubscribeTime >= int64(subscribeDuration) {
			expireTime := now + expireDuration
			err := fnSubscribeMQ(expireTime)

			if err != nil {
				me.bMQReady = false
				logger.Logf("tSaleOrderService.beginSubscribeMQ, failed, err=%v", err)

			} else {
				lastSubscribeTime = now
				me.bMQReady = true
				logger.Logf("tSaleOrderService.beginSubscribeMQ, done")
			}
		}
		time.Sleep(pauseDuration)
	}
}

func fnSubscribeMQ(expireTime int64) error {
	msg := &models.SubscribeMsg{
		ClientID: gMQClientID,
		Topic: gMQSubscribeTopic,
		NotifyUrl: gMQServerURL + PathOfNotifyStockOutbound,
		ExpireTime: expireTime,
	}
	url := gMQServerURL + "/subscribe"
	return fnPost(msg, url)
}


func fnPost(msg interface{}, url string) error {
	body,_ := json.Marshal(msg)
	rsp, err := http.Post(url, "application/json;charset=utf-8", bytes.NewReader(body))
	if err != nil {
		return err
	}

	defer rsp.Body.Close()
	j, err := ioutil.ReadAll(rsp.Body)
	if err != nil {
		return err
	}
	ok := &models.OkMsg{}
	err = json.Unmarshal(j, ok)
	if err != nil {
		return err
	}

	if !ok.OK {
		return gMQReplyFalse
	}

	return nil
}


func (me *tSaleOrderService) beginPublishMQ() {
	for {
		select {
		case msg := <- me.publishQueue :
			me.publishMQ(msg)
			break
		}
	}
}


func (me *tSaleOrderService) Get(orderID string) *SaleOrder {
	me.rwmutex.RLock()
	defer me.rwmutex.RUnlock()

	it,ok := me.orders[orderID]
	if ok {
		return it
	} else {
		return nil
	}
}

func (me *tSaleOrderService) Create(it *SaleOrder) error {
	me.rwmutex.Lock()
	defer me.rwmutex.Unlock()

	if len(me.publishQueue) >= gMQMaxQueuedMsg {
		return gMQNotAvailableError
	}
	me.orders[it.OrderID] = it
	me.publishQueue <- it

	return nil
}


func (me *tSaleOrderService) publishMQ(it *SaleOrder) {
	url := gMQServerURL + "/publish"

	j,_ := json.Marshal(it)
	msg := &models.TxMsg{
		GlobalID: it.OrderID,
		SubID: it.OrderID,
		SenderID: gMQClientID,
		Topic: gMQPublishTopic,
		CreateTime: it.CreateTime,
		Content: string(j),
	}

	for i := 0;i < gMQMaxPublishRetry;i++ {
		err := fnPost(msg, url)
		if err != nil {
			logger.Logf("tSaleOrderService.publishMQ, failed, err=%v, order=%v", err, it)
			time.Sleep(gMQPublishInterval)

		} else {
			logger.Logf("tSaleOrderService.publishMQ, done, order=%s/%v", it.OrderID, it)
			return
		}
	}

	// publish failed
	logger.Logf("tSaleOrderService.publishMQ, failed max retries, order=%v", it)
	_,_ = me.Update(it.OrderID, StatusNotDelivered, StatusMQServiceFailed)
}


func (me *tSaleOrderService) Update(orderID string, oldStatusFlag int32, newStatusFlag int32) (error, *SaleOrder) {
	me.rwmutex.RLock()
	defer me.rwmutex.RUnlock()

	it, ok := me.orders[orderID]
	if !ok {
		return gNotFoundError, nil
	}

	if !atomic.CompareAndSwapInt32(&it.StatusFlag, oldStatusFlag, newStatusFlag) {
		return gStatusChangedError, it
	}

	it.StatusFlag = newStatusFlag
	return nil, it
}

var gMQReplyFalse = errors.New("mq reply false")
var gMQNotAvailableError = errors.New("mq not ready")
var gNotFoundError = errors.New("order not found")
var gStatusChangedError = errors.New("status changed")
var gMQMaxPublishRetry = 10
var gMQPublishInterval = 1*time.Second
var gMQSubscribeTopic = "sale-order.stock.outbound"
var gMQPublishTopic = "sale-order.created"
var gMQClientID = "sale-order-service"
var gMQServerURL = "http://localhost:3333"
var gMQMaxQueuedMsg = 1024

var MockSaleOrderService = newSaleOrderService()